package com.ywps.vaas.framework.util
import com.alibaba.fastjson.JSON
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

object SqlUtil {

  /**
   * 根据mysql查询语句返回dataframe对象
   * @param sql  查询语句
   * @param spark
   * @return dataframe对象
   */
  def getDataFrameFromMysql(sql:String,spark:SparkSession):DataFrame ={
    val list: List[String] = MysqlUtils.queryBySql(sql)
    val keys: List[String] = JsonUtil.getKeysFromJsonObject(JSON.parseObject(list.head))
    var row_list:java.util.List[Row] = new java.util.ArrayList[Row]()
    list.foreach(entry => row_list.add(JsonUtil.getValuesFromJsonObject(JSON.parseObject(entry))))
    val headSeq: List[StructField] = keys.map(key => StructField(key, StringType, nullable = false)).toList
    val df: DataFrame = spark.createDataFrame(row_list, StructType(headSeq))
    df
  }

}
